package com.wugui.datax.rpc.remoting.invoker.route.impl;

import cn.hutool.core.collection.CollUtil;
import com.wugui.datax.rpc.remoting.invoker.route.XxlRpcLoadBalance;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * lru
 *
 * @author xuxueli 2018-12-04
 */
public class XxlRpcLoadBalanceLFUStrategy extends XxlRpcLoadBalance {

  private final ConcurrentMap<String, HashMap<String, Integer>> jobLfuMap =
      new ConcurrentHashMap<String, HashMap<String, Integer>>();
  private long CACHE_VALID_TIME = 0;

  public String doRoute(String serviceKey, TreeSet<String> addressSet) {

    // cache clear
    if (System.currentTimeMillis() > CACHE_VALID_TIME) {
      jobLfuMap.clear();
      CACHE_VALID_TIME = System.currentTimeMillis() + 1000 * 60 * 60 * 24;
    }

    // lfu item init
    HashMap<String, Integer> lfuItemMap =
        jobLfuMap.get(serviceKey); // Key排序可以用TreeMap+构造入参Compare；Value排序暂时只能通过ArrayList；
    if (lfuItemMap == null) {
      lfuItemMap = new HashMap<String, Integer>();
      jobLfuMap.putIfAbsent(serviceKey, lfuItemMap); // 避免重复覆盖
    }

    // put new
    for (String address : addressSet) {
      if (!lfuItemMap.containsKey(address) || lfuItemMap.get(address) > 1000000) {
        lfuItemMap.put(address, 0);
      }
    }

    // remove old
    List<String> delKeys = new ArrayList<>();
    for (String existKey : lfuItemMap.keySet()) {
      if (!addressSet.contains(existKey)) {
        delKeys.add(existKey);
      }
    }
    if (CollUtil.isNotEmpty(delKeys)) {
      for (String delKey : delKeys) {
        lfuItemMap.remove(delKey);
      }
    }

    // load least userd count address
    List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<>(lfuItemMap.entrySet());
    lfuItemList.sort(Map.Entry.comparingByValue());

    Map.Entry<String, Integer> addressItem = lfuItemList.get(0);
    String minAddress = addressItem.getKey();
    addressItem.setValue(addressItem.getValue() + 1);

    return minAddress;
  }

  @Override
  public String route(String serviceKey, TreeSet<String> addressSet) {
    return doRoute(serviceKey, addressSet);
  }
}
